package com.baeldung.libraries.debezium.config;

import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Profile;

import java.io.File;
import java.io.IOException;

@Configuration
@Profile("postgresql")
public class DebeziumPostgresqlConnectorConfig {

    @Bean
    @ConfigurationProperties("customer.postgresql.datasource")
    public SourceDatabaseProperties sourceDatabaseProperties() {
        return new SourceDatabaseProperties();
    }

    /**
     * Customer Database Connector Configuration
     */
    @Bean
    public io.debezium.config.Configuration customerPostgresqlConnector(SourceDatabaseProperties properties) throws IOException {
        File offsetStorageTempFile = new File("pg-debezium/test_db_offsets.json");
        return io.debezium.config.Configuration.create()
                .with("name", "customer-postgresql-connector")
                .with("connector.class", "io.debezium.connector.postgresql.PostgresConnector")
                .with("plugin.name", "wal2json")
                .with("offset.storage", "com.baeldung.libraries.debezium.config.JacksonFileOffsetBackingStore")
                .with("offset.storage.file.filename", offsetStorageTempFile.getAbsolutePath())
                .with("offset.flush.interval.ms", "60000")
                .with("database.hostname", properties.getHost())
                .with("database.port", properties.getPort())
                .with("database.user", properties.getUsername())
                .with("database.password", properties.getPassword())
                .with("database.dbname", properties.getDatabase())
//                .with("database.include.list", properties.getDatabase())
                .with("table.include.list", properties.getTables())
                .with("include.schema.changes", "false")
                .with("database.allowPublicKeyRetrieval", "true")
                .with("database.server.name", "customer-postgresql-db-server")
//                .with("converter.schemas.enable", "false")
                .build();
    }
}
